//
// Created by 郝羽 on 2018/6/25.
//

#include <glib-unix.h>
#include "child.hpp"

#include "conn.hpp"
#include "config.hpp"
#include "metric.hpp"
#include "metric_client.hpp"
#include "ping.hpp"

constexpr std::chrono::milliseconds kTimeoutInterval(20);
constexpr std::chrono::milliseconds kMaxTimeoutInterval(200);
constexpr std::chrono::milliseconds kStatInterval(1000);

struct Child {
    GMainLoop* loop;
    Context ctx;
    std::vector<Conn*> conns;
    Metric cur_metric;
    Clock::time_point start_time;
    Clock::time_point next_timeout;
    Clock::time_point next_report;
    MetricClient* mclient;
    bool exiting;
    Pinger* pinger;
};

void child_handle_connection_init(Conn* conn, bool ok, void* data) {
    auto c = static_cast<Child*>(data);

    if (!ok) {
        return;
    }

    if (c->ctx.sequence == Context::Sequence::Offerer) {
        conn->trickle();
    } else if (c->ctx.sequence == Context::Sequence::Answerer) {
        conn->answer();
    }
}

void child_handle_connection_subscribe(Conn* conn, bool ok, void* data) {
    auto c = static_cast<Child*>(data);

    if (!ok) {
        conn->terminate();
        return;
    }
}

void child_handle_connection_trickle(Conn* conn, bool ok, void* data) {
    auto c = static_cast<Child*>(data);

    if (!ok) {
        conn->terminate();
        return;
    }

    conn->subscribe();
}

void child_handle_connection_terminate(Conn* conn, bool ok, void* data) {
    auto c = static_cast<Child*>(data);
    //c->conns.erase(conn->local_rtc_id());
    //delete conn;
    // 退出再删除
}

void child_handle_connection_answer(Conn* conn, bool ok, void* data) {
    auto c = static_cast<Child*>(data);

    if (!ok) {
        conn->terminate();
        return;
    }

    dbgi("answer connection success");
    conn->trickle();
}

void child_handle_connection_setup(Conn* conn, int setup_time, void* data) {
    auto c = static_cast<Child*>(data);
    c->cur_metric.setup_time.Update(setup_time);
}

ConnectionEventHandler child_conn_handler {
        .on_conn_init = child_handle_connection_init,
        .on_conn_subscribe = child_handle_connection_subscribe,
        .on_conn_trickle = child_handle_connection_trickle,
        .on_conn_terminate = child_handle_connection_terminate,
        .on_conn_answer = child_handle_connection_answer,
        .on_conn_setup = child_handle_connection_setup,
};

void child_init_rtc(Child* c, Clock::duration elapsed) {
    auto msecs = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed)
            .count();
    auto targetConnectionCount = static_cast<double>(c->ctx.concurrency)
            / c->ctx.rampUpSeconds * msecs / 1000;

    if (targetConnectionCount > c->ctx.concurrency) {
        targetConnectionCount = c->ctx.concurrency;
    }

    while (c->conns.size() < targetConnectionCount) {
        int rtc_id = c->conns.size();
        Conn* conn = nullptr;
        if (c->ctx.pub_rtc_ids.size() > 0) {
            const char* pub_rtc_id = c->ctx.pub_rtc_ids[rtc_id % c->ctx.pub_rtc_ids.size()];
            conn = new Conn(c->ctx, rtc_id, child_conn_handler, c);
            conn->set_pub_rtc(pub_rtc_id);
        } else {
            conn = new Conn(c->ctx, rtc_id, child_conn_handler, c);
        }
        conn->init();
        c->conns.emplace_back(conn);
    }
}

static gboolean child_handle_timeout(gpointer data) {
    auto c = static_cast<Child*>(data);

    auto now = Clock::now();

    auto now_epoch_ms = epochMs(now);

    auto elapsed = now - c->start_time;

    auto secs = std::chrono::duration_cast<std::chrono::seconds>(elapsed)
            .count();

    auto t = now - c->next_timeout;
    if (t > kMaxTimeoutInterval) {
        g_warning("child %d: timer drifted %d ms", getpid(),
                std::chrono::duration_cast<std::chrono::milliseconds>(
                t).count());
    }

    c->next_timeout = now + kTimeoutInterval;

    if (secs > c->ctx.durationSeconds || c->exiting) {
        for (auto& conn : c->conns) {
            if (conn->is_terminating()) {
                continue;
            }
            g_message("child %d: terminate %s", getpid(),
                    conn->local_rtc_id().c_str());
            conn->terminate();
        }

        bool all_terminated = true;

        for (auto& conn : c->conns) {
            if (!conn->is_terminated()) {
                all_terminated = false;
                break;
            }
        }

        if (all_terminated) {
            g_message("child %d: all conn terminated", getpid());
            for (auto& conn : c->conns) {
                delete(conn);
            }
            c->conns.clear();
            g_main_loop_quit(c->loop);
            return G_SOURCE_REMOVE;
        }

        return G_SOURCE_CONTINUE;
    }

    child_init_rtc(c, elapsed);

    // send packet if in symmetric mode
    // in asymmetric mode, just receive
    bool is_symmetric = c->ctx.pub_rtc_ids.empty();
    if (is_symmetric) {
        for (auto& conn : c->conns) {
            if (conn->is_setup() && !conn->is_terminating()) {
                conn->send_packets();
            }
        }
    }

    if (now >= c->next_report) {
        c->cur_metric.time = secs;

        c->cur_metric.ping_time.Update(pinger_result(c->pinger));

        for (auto& conn : c->conns) {
            auto m = conn->metric();
            c->cur_metric.rtt.Update(m.rtt);
            c->cur_metric.jitter.Update(m.jitter);
            c->cur_metric.inited.Update(1);

            if (conn->is_setup() && !conn->is_terminating()) {
                //c->cur_metric.setup_time.Update(m.setup_time);
                c->cur_metric.setup.Update(1);
                uint64_t send_bytes = 0;
                uint64_t recv_bytes = 0;
                conn->get_send_recv_bytes(send_bytes, recv_bytes);
                c->cur_metric.send.Update(send_bytes);
                c->cur_metric.recv.Update(recv_bytes);
                uint64_t send_pkts = 0;
                uint64_t recv_pkts = 0;
                conn->get_send_recv_packets(send_pkts, recv_pkts);
                c->cur_metric.send_packet.Update(send_pkts);
                c->cur_metric.recv_packet.Update(recv_pkts);
            }

            m.rtt.Clear();
            m.jitter.Clear();
            m.setup_time.Clear();
        }

        metric_client_send_metric(c->mclient, c->cur_metric);
        //child_emit_metric(g_state.current_metric);

        c->cur_metric.rtt.Clear();
        c->cur_metric.jitter.Clear();
        c->cur_metric.setup_time.Clear();
        c->cur_metric.send.Clear();
        c->cur_metric.recv.Clear();
        c->cur_metric.send_packet.Clear();
        c->cur_metric.recv_packet.Clear();
        c->cur_metric.inited.Clear();
        c->cur_metric.setup.Clear();
        c->cur_metric.ping_time.Clear();

        c->next_report += kStatInterval;
    }
    return G_SOURCE_CONTINUE;
}

gboolean child_handle_signal(gpointer data) {
    auto c = static_cast<Child*>(data);
    g_message("child received signal");
    c->exiting = true;
    return G_SOURCE_REMOVE;
}

void child_main(Context& ctx) {
    g_message("child %d: start", getpid());

    auto c = new Child();

    c->ctx = ctx;

    c->loop = g_main_loop_new(nullptr, FALSE);

    GSource* timer = g_timeout_source_new(
            std::chrono::duration_cast<std::chrono::milliseconds>(
            kTimeoutInterval).count());

    c->start_time = Clock::now();

    c->next_timeout = c->start_time + kTimeoutInterval;

    c->next_report = c->start_time + kStatInterval;

    c->exiting = false;

    g_unix_signal_add(SIGINT, child_handle_signal, c);

    g_source_set_callback(timer, child_handle_timeout, c, nullptr);

    g_source_attach(timer, nullptr);

    c->mclient = metric_client_open(c->ctx.metric_server_address);

    c->pinger = pinger_open(c->ctx);

    g_main_loop_run(c->loop);

    g_main_loop_unref(c->loop);

    g_source_destroy(timer);

    g_source_unref(timer);

    //g_main_context_pop_thread_default(g_main_context_default());

    g_message("child: %d exit", getpid());

    if (c->mclient != nullptr) {
        metric_client_close(c->mclient);
        c->mclient = nullptr;
    }

    if (c->pinger != nullptr) {
        pinger_close(c->pinger);
        c->pinger = nullptr;
    }

    delete c;

    return;
}